package cn.xuexiyuan.flinkstudy.checkpoint;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;

/**
 * @Description:
 * @Author 左龙龙
 * @Date 21-3-29
 * @Version 1.0
 **/
public class CheckpointDemo01 {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 设置并行度1方便观察
        env.setParallelism(1);
        // 以下 checkpoint 相关配置
        // 每隔 1s 执行异常 checkpoint
        env.enableCheckpointing(1000);
        // 设置 State 状态存储介质
        // 1. Memory : State 存内存， Checkpoint 存内存 -- 开发使用
        // 2. Fs : State 存内容， Checkpoint 存 FS（本地/HDFS） -- 一般情况下使用
        // 3. RocksDB : State 存 RocksDB（内存+磁盘）， Checkpoint 存 FS（本地+磁盘） —— 超大状态使用,但是对于状态的读写效率要低一点
        env.setStateBackend(new FsStateBackend("file:///tmp/ckp"));

        // ====== 类型2： 建议参数 ======
        // 设置两个 Checkpoint 之间最少等待时间， 默认是 0，如设置 CHeckpoint 直接最少要等待 500ms (为了避免每隔 1000ms 做一次Checkpoint的时候，前一次太慢和后一次重叠到一起了)
        // 如： 高速公路上，每隔 1s 关口放行一辆车，但是规定了两车之间的最小车距为 500m
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        // 表示容忍检查点失败的次数, 默认值 0 表示不容忍任何检查点失败
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(10);
        // CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION   当作业被取消时，删除外部的 checkpoint
        // CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION   当作业被取消时，保留外部的 checkpoint
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // ====== 类型3： 保持默认即可 =====
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 设置 checkpoint 的超时时间，如果checkpoint 在 60s 馁未完成说明该次 Checkpoint 失败，则丢弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 设置同一时间有多少个 Checkpoint 可以同时执行， 默认值 1
        // env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);


        // 1.source
        DataStreamSource<String> ds1 = env.socketTextStream("localhost", 6666);


        // 2.transformation
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = ds1.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (s, collector) -> {
            for (String token : s.split("\\W+")) {
                collector.collect(Tuple2.of(token, 1));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT));
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordAndOne.keyBy(d -> d.f0).sum(1);


        // 3.sink
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("localhost").build();
        RedisSink<Tuple2<String, Integer>> redisSink = new RedisSink(conf, new MyRedisMapper());
        result.addSink(redisSink);


        // 4.excute
        env.execute();
    }

}
